package com.ghr.utils;

import java.io.IOException;
import java.lang.reflect.Type;
import java.util.Date;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;

import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.core.CountRequest;
import org.elasticsearch.client.core.CountResponse;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.CreateIndexResponse;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.core.TimeValue;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.UpdateByQueryRequest;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.script.Script;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;

import static com.ghr.common.constant.ElasticSearchConstant.ES_PAGE_SIZE;
import static com.ghr.common.constant.ElasticSearchConstant.ES_TIME_OUT;

@Component
@Slf4j
public class ElasticSearchUtil<T> {

    @Autowired
    private RestHighLevelClient restHighLevelClient;

    /**
     * 创建索引
     *
     * @param indexName
     * @return
     * @throws Exception
     */
    public boolean createIndex(String indexName) {
        if (isIndexExist(indexName)) {
            log.error("Index is  exits. index name = {}", indexName);
            return false;
        }
        try {
            CreateIndexRequest request = new CreateIndexRequest(indexName);
            CreateIndexResponse response = null;
            response = restHighLevelClient.indices()
                    .create(request, RequestOptions.DEFAULT);
            log.info("createIndex response = {}", response);
            restHighLevelClient.close(); // 关闭资源
            return response.isAcknowledged();
        } catch (Exception e) {
            log.error("createIndex error. e = {}", e);
            return false;
        }

    }

    /**
     * 判断索引是否存在
     *
     * @param indexName
     * @return
     * @throws Exception
     */
    public boolean isIndexExist(String indexName) {
        boolean exists = false;
        try {
            GetIndexRequest request = new GetIndexRequest(indexName);
            exists = restHighLevelClient.indices().exists(request, RequestOptions.DEFAULT);
            log.info("index {} is {}", indexName, exists);
            restHighLevelClient.close(); // 关闭资源
        } catch (Exception e) {
            log.error("isIndexExist error. e = {}", e);
        }
        return exists;
    }

    /**
     * 添加数据
     *
     * @param object    新增数据
     * @param indexName 索引名称
     * @param objectId  数据id
     * @return
     */
    public boolean addDocument(Object object, String indexName, String objectId) {
        try {
            log.info("addDocument params[object:{}, indexName:{}, objectId:{}]", object, indexName, objectId);
            IndexRequest indexRequest = new IndexRequest(indexName);
            if (!StringUtils.isEmpty(objectId)) {
                indexRequest.id(objectId);
            }
            byte[] json = JSON.toJSONBytes(object);
            indexRequest.source(json, XContentType.JSON);
            IndexResponse response = restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);
            log.info("add Document = {}, id = {}, response = {}", object, objectId, response);
//            restHighLevelClient.close(); // 关闭资源
            return response.status().getStatus() == RestStatus.CREATED.getStatus();
        } catch (Exception e) {
            log.info("add Document error. e = {}", e);
            return false;
        }
    }

    /**
     * 批量插入
     *
     * @param objectList
     * @param indexName
     * @return
     */
    public boolean addObjectList(List<T> objectList, String indexName) {

        IPage<T> tmpPage = new Page<>();
        tmpPage.setTotal(objectList.size());
        long pageSize = ES_PAGE_SIZE;
        tmpPage.setSize(pageSize);
        for (int curPage = 0; curPage < tmpPage.getPages(); curPage++) {
            try {
                List<?> tmpCollect = objectList.stream().skip(curPage * pageSize)
                        .limit(pageSize).collect(Collectors.toList());
                BulkRequest bulkRequest = new BulkRequest();
                bulkRequest.timeout(TimeValue.timeValueMillis(ES_TIME_OUT));
                Iterator iterator = tmpCollect.iterator();
                while (iterator.hasNext()) {
                    Object tmpObject = iterator.next();
                    String json = JSON.toJSONString(tmpObject);
                    bulkRequest.add(new IndexRequest(indexName).source(json, XContentType.JSON));
                }

                BulkResponse bulkItemResponses = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
                if (bulkItemResponses.hasFailures()) {
                    log.error("buldResponse error. {}", bulkItemResponses.buildFailureMessage());
                }
            } catch (Exception e) {
                log.error("addObjectList catch exception : {}", e);
            }
        }
        return true;
    }

    /**
     * 通过id删除index中的数据
     *
     * @param indexName
     * @param id
     * @return
     */
    public boolean deleteObjectById(String indexName, String id) {
        try {
            DeleteRequest deleteRequest = new DeleteRequest(indexName, id);
            DeleteResponse response = restHighLevelClient.delete(deleteRequest, RequestOptions.DEFAULT);
            log.info("deleteObjectById success. index = {}, id = {}, response = {}", indexName, id, response);
            restHighLevelClient.close(); // 关闭资源
            return response.status().equals(RestStatus.OK);
        } catch (Exception e) {
            log.error("deleteObjectById error. e = {}", e);
            return false;
        }
    }

    /**
     * 根据id更新对象
     *
     * @param indexName
     * @param id
     * @param object
     * @return
     */
    public boolean updateObjectById(String indexName, String id, Object object) {
        try {
            UpdateRequest updateRequest = new UpdateRequest(indexName, id);
            String json = JSON.toJSONString(object);
            updateRequest.doc(json, XContentType.JSON);
            UpdateResponse updateResponse = restHighLevelClient.update(updateRequest, RequestOptions.DEFAULT);
            log.info("updateObjectById content:{}", json);
            log.info("updateObjectById success. id = {}, response = {}", id, updateResponse);
            restHighLevelClient.close(); // 关闭资源
            return updateResponse.status().equals(RestStatus.OK);
        } catch (Exception e) {
            log.error("updateObjectById error. id = {}, e = {}", id, e);
            return false;
        }
    }

    /**
     * 根据id更新多个对象
     *
     * @param indexName
     * @param id
     * @param map
     * @return
     */
    public boolean updateObjectById(String indexName, String id, Map<String, Object> map) {
        try {
            UpdateRequest updateRequest = new UpdateRequest(indexName, id);
            updateRequest.doc(map, XContentType.JSON);
            UpdateResponse updateResponse = restHighLevelClient.update(updateRequest, RequestOptions.DEFAULT);
            log.info("id:{}, updateObjectById status:{}", id, updateResponse.status());
            restHighLevelClient.close(); // 关闭资源
            return updateResponse.status().equals(RestStatus.OK);
        } catch (Exception e) {
            log.error("updateObjectById error. id = {}, e = {}", id, e);
            return false;
        }
    }

    /**
     * 根据查询条件并使用script脚本更新
     *
     * @param queryBuilder
     * @param indexName
     * @param updateFieldMap
     * @return
     */
    public boolean updateObjectByQuery(QueryBuilder queryBuilder, String indexName,
                                       Map<String, Object> updateFieldMap) {
        try {
            UpdateByQueryRequest updateByQueryRequest = new UpdateByQueryRequest(indexName);
            updateByQueryRequest.setQuery(queryBuilder);
            StringBuilder scriptBuilder = new StringBuilder();
            Set<String> keys = updateFieldMap.keySet();
            for (String key : keys) {
                String appendValue = "";
                Object valueObject = updateFieldMap.get(key);
                if (valueObject == null) {
                    continue;
                }
                if (valueObject instanceof Number) {
                    appendValue = valueObject.toString();
                } else if (valueObject instanceof String) {
                    if (!StringUtils.isEmpty(valueObject.toString())) {
                        appendValue = "'" + valueObject.toString() + "'";
                    }
                } else if (valueObject instanceof Date) {
                    String dateStr = DateUtil.getDateStr("yyyy-MM-dd HH:mm:ss", (Date) valueObject);
                    appendValue = "'" + dateStr + "'";
                } else {
                    appendValue = valueObject.toString();
                }
                if (!StringUtils.isEmpty(appendValue)) {
                    scriptBuilder.append("ctx._source.").append(key).append("=").append(appendValue).append(";");
                }
            }
            String scriptStr = scriptBuilder.toString();
            updateByQueryRequest.setScript(new Script(scriptStr));
            BulkByScrollResponse response = restHighLevelClient.updateByQuery(updateByQueryRequest,
                    RequestOptions.DEFAULT);
            log.info("updateObjectByQuery response = {}", response);
            restHighLevelClient.close(); // 关闭资源
            return true;
        } catch (Exception e) {
            return false;
        }
    }

    /**
     * 根据id获取es存储对象
     *
     * @param indexName
     * @param id
     * @param clazz
     * @param <T>
     * @return
     */
    public <T> T getObjectById(String indexName, String id, Type clazz) {
        try {
            GetRequest getRequest = new GetRequest(indexName, id);
            GetResponse getResponse = restHighLevelClient.get(getRequest, RequestOptions.DEFAULT);
            if (getResponse.isExists()) {
                T value = JSON.parseObject(getResponse.getSourceAsBytes(), clazz);
//                T value = JSON.parseObject(getResponse.getSourceAsString(), clazz);
                log.info("getObjectById indexname = {}, id = {}, result = {}", indexName, id, value);
                restHighLevelClient.close(); // 关闭资源
                return value;
            }
        } catch (Exception e) {
            log.error("getObjectById catch error. e = {}", e);
        }
        return null;
    }

    /**
     * 获取查询数量
     *
     * @param sourceBuilder
     * @param indexName
     * @param type
     * @return
     */
    public long getListCount(SearchSourceBuilder sourceBuilder, String indexName, String type) throws IOException {
        CountResponse countResponse = null;
        try {
            if (sourceBuilder == null || StringUtils.isEmpty(indexName)) {
                return 0;
            }
            CountRequest countRequest = new CountRequest(indexName);
            if (!StringUtils.isEmpty(type)) {
                countRequest.types(type);
            }
            countRequest.source(sourceBuilder);
            countResponse = restHighLevelClient.count(countRequest, RequestOptions.DEFAULT);
        } catch (Exception e) {
            log.error("getListCount catch error. e = {}", e);
            return 0;
        }

        if (countResponse == null || countResponse.status().getStatus() != RestStatus.OK.getStatus()) {
            restHighLevelClient.close(); // 关闭资源
            return 0;
        } else {
            restHighLevelClient.close(); // 关闭资源
            return countResponse.getCount();
        }
    }

    /*
        轻量级的返回SearchResponse的查询工具
     */
    public SearchResponse getSearchResponse(QueryBuilder queryBuilder, String index) throws IOException {
        // 开始查找
        SearchRequest searchRequest = new SearchRequest();
        searchRequest.indices(index);
        // 创建查询的请求体
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        sourceBuilder.query(queryBuilder);
        searchRequest.source(sourceBuilder);

        SearchResponse searchResponse = null;
        try {
            searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
        } catch (IOException e) {
            e.printStackTrace();
        }
        restHighLevelClient.close(); // 关闭资源
        return searchResponse;
    }
}
